package cn.itcast.kafka.optimize.manual;
import cn.itcast.kafka.config.KafkaConsumerConfig;
import cn.itcast.kafka.utils.DeadLetterQueueUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

public class KafkaConsumerManualCommitManualSubmit {
    // 最大重试次数
    private static final int MAX_RETRY_COUNT = 3;
    // 用于记录每个消息的重试次数（key: 分区+offset，value: 重试次数）
    private static final Map<String, Integer> retryCountMap = new ConcurrentHashMap<>();

    public static void main(String[] args) {
        // 获取消费者配置
        Properties props = KafkaConsumerConfig.getConsumerProperties();
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        // 订阅主题
        consumer.subscribe(Arrays.asList("test1"));

        try {
            while (true) {
                // 拉取消息
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(5));
                if (records.isEmpty()) {
                    continue;
                }

                // 用于记录已处理成功的消息offset
                Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();

                for (ConsumerRecord<String, String> record : records) {
                    String topic = record.topic();
                    int partition = record.partition();
                    long offset = record.offset();
                    String key = record.key();
                    String value = record.value();

                    // 生成唯一标识（用于重试计数）
                    String recordKey = topic + "-" + partition + "-" + offset;
                    // 获取当前重试次数（默认0）
                    int retryCount = retryCountMap.getOrDefault(recordKey, 0);

                    boolean processSuccess = false;
                    try {
                        // 处理消息（实际业务逻辑）
                        processMessage(record);
                        processSuccess = true;
                        // 处理成功，移除重试计数
                        retryCountMap.remove(recordKey);
                        System.out.printf("消息处理成功: topic=%s, partition=%d, offset=%d%n",
                                topic, partition, offset);
                    } catch (Exception e) {
                        System.err.printf("消息处理失败（第%d次）: topic=%s, partition=%d, offset=%d, 错误: %s%n",
                                retryCount + 1, topic, partition, offset, e.getMessage());

                        if (retryCount < MAX_RETRY_COUNT - 1) {
                            // 未达最大重试次数，记录重试次数，等待下次poll时重试
                            retryCountMap.put(recordKey, retryCount + 1);
                            // 休眠一段时间再重试（指数退避策略）
                            Thread.sleep(1000 * (retryCount + 1));
                        } else {
                            // 达到最大重试次数，发送到死信队列
                            DeadLetterQueueUtil.sendToDLQ(topic, key, value, e);
                            // 移除重试计数（已进入死信队列）
                            retryCountMap.remove(recordKey);
                            processSuccess = true; // 死信处理后标记为"处理完成"
                        }
                    }

                    // 处理成功（包括进入死信队列的情况），记录offset准备提交
                    if (processSuccess) {
                        TopicPartition topicPartition = new TopicPartition(topic, partition);
                        // offset+1表示下一条要消费的消息
                        offsetsToCommit.put(topicPartition, new OffsetAndMetadata(offset + 1));
                    }
                }

                // 批量提交已处理成功的消息offset
                if (!offsetsToCommit.isEmpty()) {
                    consumer.commitSync(offsetsToCommit);
                    System.out.println("已提交offset: " + offsetsToCommit);
                }
            }
        } catch (Exception e) {
            System.err.println("消费者发生异常: " + e.getMessage());
            e.printStackTrace();
        } finally {
            // 关闭资源
            consumer.close();
            DeadLetterQueueUtil.close();
        }
    }

    /**
     * 消息处理逻辑（模拟业务处理）
     */
    private static void processMessage(ConsumerRecord<String, String> record) throws Exception {
        String value = record.value();
        // 模拟处理异常（例如：值为偶数时抛出异常）
        if (value != null && value.matches("\\d+") && Integer.parseInt(value) % 2 == 0) {
            throw new RuntimeException("处理偶数消息时发生异常（模拟）");
        }
        // 实际业务处理逻辑...
    }
}
